package com.chencong.online.driver.dwd;

import com.chencong.online.bean.UniqueVisitorBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.UVBloomFilterProcessFunc;
import com.chencong.online.function.UVTriggerFunc;
import com.chencong.online.function.UserBehaviorBeanMapFunc;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.junit.Test;

import java.net.URL;
import java.time.Duration;
import java.time.LocalDateTime;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdUVWithBloomFilterDriver
 * @description:布隆过滤器优化求UV
 * @author: chencong
 * @create: 2021-12-28 16:12
 **/
public class DwdUVWithBloomFilterDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //todo 1 source
        URL resource = DwdPageViewDriver.class.getResource("/UserBehavior.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print("获取原始数据");
        //todo 2 transform
        //转换层javabean并提取时间戳
        SingleOutputStreamOperator<UserBehaviorBean> userBehaviorDS = inputDS
                .map(new UserBehaviorBeanMapFunc())
                .filter(data -> "pv".equals(data.getBehavior()))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehaviorBean>() {
                            @Override
                            public long extractTimestamp(UserBehaviorBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        }));
        //开窗去重求UV
        SingleOutputStreamOperator<UniqueVisitorBean> resultDS = userBehaviorDS
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
                .trigger(new UVTriggerFunc()) //因为是来一条数据判断一条数据，因此不会注册定时，需要自定义触发器
                .process(new UVBloomFilterProcessFunc());//自定义布隆过滤器

        resultDS.print("最终结果");
        env.execute("DwdUVWithBloomFilterDriver");
    }
}
